package com.gugee.ins.data.task.blogger.service;

import com.gugee.common.task.dto.TaskParam;
import com.gugee.common.task.model.TaskRecord;
import com.gugee.common.task.service.AbstractProducerHandler;
import com.gugee.ins.common.dal.service.blogger.InsBloggerMainDao;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.gugee.ins.data.task.config.TaskConfig;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

/**
 * 大博主及粉丝地理位置采集任务
 * @author Huangxin
 */
@Component
@Slf4j
public class InsBloggerLocationCollectSvc extends AbstractProducerHandler<InsBloggerMessage> {
    /**
     *
     */
    @Autowired
    @Qualifier("insBloggerMsgBaseInfoProducer")
    private KafkaTemplate<Long, InsBloggerMessage> insBloggerMsgBaseInfoProducer;

    /**
     *
     */
    @Autowired
    public InsBloggerMainDao insBloggerMainDao;

    /**
     *
     */
    @Autowired
    public TaskConfig taskConfig;

    /**
     *
     * @param taskRecord 业务类型数据的采集页
     * @param taskParam  任务参数，包含了每页的页码大小
     * @return
     */
    @Override
    public List<InsBloggerMessage> findMessageObject(TaskRecord taskRecord, TaskParam taskParam) {
        // 开始id
        long startId = (taskRecord.getCurrentPage() - 1) * taskParam.getRecords() + 1L;
        // 结束的id
        long endId = (long)taskRecord.getCurrentPage() * taskParam.getRecords();
        // 需要采集的数据
        List<InsBloggerMessage> rst = insBloggerMainDao.findInsLocationCollectMessageByPages(startId,endId,
                taskConfig.getInsBloggerMainFollowerCountMin());
        return rst == null ? new ArrayList<>() : rst;
    }

    /**
     * 查询当前有最大BloggerMain的id值，用于代表记录数
     * @return
     */
    @Override
    public long objectTotals() {
        return insBloggerMainDao.getMaxId();
    }

    /**
     *
     * @param messages
     * @param topic
     */
    @Override
    public void sendMsg(List<InsBloggerMessage> messages, String topic) {
        messages.forEach(msg->{
            msg.setRetryTimes(taskConfig.getInsLocationRetryTimes());
            msg.setRetryLimitTimes(taskConfig.getInsLocationRetryTimes());
            insBloggerMsgBaseInfoProducer.send(taskConfig.getInsBloggerLocationOnlineTopic(),msg);
        });
        XxlJobLogger.log("投递主题={},投递消息数={}", taskConfig.getInsBloggerLocationOnlineTopic(),messages.size());
        log.info("投递主题={},投递消息数={}", taskConfig.getInsBloggerLocationOnlineTopic(),messages.size());
    }
}
